package com.bijava.service.impl;

import com.bijava.config.TaskJobListener;
import com.bijava.model.TaskJobVO;
import com.bijava.service.ElasticJobService;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * @author bijava
 * @version 1.0
 * @since 20-8-21
 */
@Service
public class ElasticJobServiceImpl implements ElasticJobService {
    @Resource
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    public void addSimpleJob(TaskJobVO taskJobVO, SimpleJob simpleJob) {
        // 配置任务（时间事件、定时任务业务逻辑、调度器）
        JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder(
                taskJobVO.getJobName(), taskJobVO.getJobCron(), taskJobVO.getShardCount())
                .shardingItemParameters(taskJobVO.getShardItem())
                .jobParameter(taskJobVO.getJobName())
                .description("simple类型作业")
                .build();

        JobTypeConfiguration jobTypeConfiguration = new SimpleJobConfiguration(
                jobCoreConfiguration, simpleJob.getClass().getCanonicalName());

        //overwrite-本地配置是否可覆盖注册中心配置
        //disabled-作业是否禁止启动,可用于部署作业时，先禁止启动，部署结束后统一启动
        //作业监控端口,建议配置作业监控端口, 方便开发者dump作业信息。使用方法: echo “dump” | nc 127.0.0.1 9888
        LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration
                .newBuilder(jobTypeConfiguration)
                .monitorPort(9888)
                .overwrite(true)
                .disabled(false)
                .build();

        TaskJobListener taskJobListener = new TaskJobListener();
        // 加载执行
        SpringJobScheduler jobScheduler = new SpringJobScheduler(
                simpleJob, zookeeperRegistryCenter,
                liteJobConfiguration, taskJobListener);
        jobScheduler.init();
    }

    @Override
    public void addDataFlowJob(TaskJobVO taskJobVO, DataflowJob dataflowJob) {
        // 配置任务（时间事件、定时任务业务逻辑、调度器）
        JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder(
                taskJobVO.getJobName(), taskJobVO.getJobCron(), taskJobVO.getShardCount())
                .shardingItemParameters(taskJobVO.getShardItem())
                .jobParameter(taskJobVO.getJobName())
                .description("dataFlow类型作业")
                .build();

        boolean streamingProcess = true;
        JobTypeConfiguration jobTypeConfiguration = new DataflowJobConfiguration(
                jobCoreConfiguration, dataflowJob.getClass().getCanonicalName(),streamingProcess);

        //overwrite-本地配置是否可覆盖注册中心配置
        //disabled-作业是否禁止启动,可用于部署作业时，先禁止启动，部署结束后统一启动
        //作业监控端口,建议配置作业监控端口, 方便开发者dump作业信息。使用方法: echo “dump” | nc 127.0.0.1 9888
        LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration
                .newBuilder(jobTypeConfiguration)
                .monitorPort(9888)
                .overwrite(true)
                .disabled(false)
                .build();

        TaskJobListener taskJobListener = new TaskJobListener();
        // 加载执行
        SpringJobScheduler jobScheduler = new SpringJobScheduler(
                dataflowJob, zookeeperRegistryCenter,
                liteJobConfiguration, taskJobListener);
        jobScheduler.init();
    }
}